package biz.ws.ds;

import api.common.util.ListUtils;
import biz.ws.ws.Message;
import biz.ws.ws.MsgDataSource;
import biz.ws.ws.WsCst;
import com.mongodb.client.result.UpdateResult;
import org.springframework.beans.factory.DisposableBean;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.data.mongodb.core.MongoTemplate;
import org.springframework.data.mongodb.core.query.Criteria;
import org.springframework.data.mongodb.core.query.Query;
import org.springframework.data.mongodb.core.query.Update;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.time.LocalDateTime;
import java.util.List;
import java.util.concurrent.BlockingQueue;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;
import java.util.stream.Collectors;

/**
 * @创建人 wsdc
 * @时间 2021/4/3
 * @描述 $
 */
@Component
public class MongoMsgDataSource implements MsgDataSource, DisposableBean {
    @Autowired
    MongoTemplate  mongo;

    private ThreadPoolExecutor tpe = new ThreadPoolExecutor(4,4,30, TimeUnit.SECONDS,new LinkedBlockingQueue<>());

    @Override
    public void save(Message msg) {
        tpe.execute(() -> {
            //  可能会重复插入 ???? 如果消息重复推送的话
            mongo.save(msg);
        });
    }

    @Override
    public void update(Message msg) {
        tpe.execute(() -> {
            Query query = Query.query(Criteria.where("id").is(msg.getId()));
            mongo.update(Message.class)
                    .matching(query)
                    .apply(Update.update("status",msg.getStatus()))
            ;
        });
    }

    @Override
    public void update(List<Message> msgList) {
        tpe.execute(() -> {
            if(ListUtils.isEmpty(msgList)){
                return;
            }

            List<Long> collect = msgList.stream()
                    .map(Message::getId)
                    .collect(Collectors.toList());

            Query query = Query.query(Criteria.where("_id").in(collect));
            UpdateResult status = mongo.update(Message.class)
                    .matching(query)
                    .apply(Update.update("status", WsCst.MsgStatus.SEND))
                    .all();
            int k = 0;
        });
    }

    @Override
    public void loadUnreadMsg(LocalDateTime after,Long toId, Consumer<List<Message>> consumer) {
        tpe.execute(() -> {
            //  仅仅是发送给我的消息
            Query query = Query.query(Criteria.where("toId").is(toId)
                    .and("createTime").gte(after)
                    .and("status").in(WsCst.MsgStatus.NOT_SEND,WsCst.MsgStatus.SEND_FAILURE)
            );
            List<Message> messages = mongo.find(query, Message.class);
            consumer.accept(messages);
        });

    }

    @Override
    public void historyMsg(LocalDateTime after,Long toId, Consumer<List<Message>> consumer) {
        tpe.execute(() -> {
            //  和我相关的消息
            Query query = Query.query(Criteria.where("toId").is(toId)
                    .orOperator(Criteria.where("fromId").is(toId))
                    .and("createTime").gte(after)
            );

            List<Message> messages = mongo.find(query, Message.class);
            consumer.accept(messages);
        });
    }

    @Override
    public void destroy() throws Exception {
        tpe.shutdown();
    }
}
